package com.cetc.sdp.kmga.cs.util;

import org.w3c.dom.Document;
import org.w3c.dom.Element;
import org.w3c.dom.Node;
import org.w3c.dom.NodeList;
import org.xml.sax.SAXException;

import javax.xml.parsers.DocumentBuilder;
import javax.xml.parsers.DocumentBuilderFactory;
import javax.xml.parsers.ParserConfigurationException;
import java.io.IOException;
import java.io.InputStream;
import java.util.*;
import java.util.function.Consumer;
import java.util.regex.Pattern;

/**
 * @description:
 * @author： DengQiang
 * @date: 2018/3/12 15:08
 */
public class StreamJobConfImpl implements StreamJobConf {

    private Map<String, String> streamProperties = new HashMap<>(32);
    private Map<String, Object> kafkaProperties = new HashMap<>(32);
    private Map<String, Object> hbaseProperties = new HashMap<>(32);
    private Map<String, String> rabbitMqProperties = new HashMap<>(32);
    private Map<Integer, HTableFiled> logFields = new HashMap<>(32);

    private static final String ROOT_ELEMENT_NAME= "application";
    private static final String STREAM_ELEMENT_NAME= "stream-config";
    private static final String KAFKA_ELEMENT_NAME= "kafka-consumer-config";
    private static final String HBASE_ELEMENT_NAME= "hbase-client-config";
    private static final String RABBITMQ_ELEMENT_NAME= "rabbitmq-consumer-config";

    public StreamJobConfImpl(String configFile) {
        parse(configFile);
    }

    private static Pattern COMMA = Pattern.compile(",");

    private void parse(String config) {
        DocumentBuilderFactory docBuilderFactory
                = DocumentBuilderFactory.newInstance();
        docBuilderFactory.setIgnoringComments(true);
        InputStream is = StreamingConfiguration.class.getClassLoader().getResourceAsStream(config);
        if (is == null) {
            throw new NullPointerException("配置文件不存在: " + config);
        }
        try {
            DocumentBuilder domBuiler = docBuilderFactory.newDocumentBuilder();
            Document doc = domBuiler.parse(is);

            Consumer<Node> parseStreamNode = (parent) -> {
                for (Node prop = parent.getFirstChild(); prop != null; prop = prop.getNextSibling()) {
                    NodeList nodeList = prop.getChildNodes();
                    String propName = null;
                    String propValue = null;
                    for (int i = 0; i < nodeList.getLength(); i++) {
                        Node n = nodeList.item(i);
                        if (n.getNodeType() == Node.ELEMENT_NODE) {
                            if ("name".equals(n.getNodeName())) {
                                propName = n.getFirstChild().getNodeValue();
                            } else if ("value".equals(n.getNodeName())) {
                                propValue = n.getFirstChild().getNodeValue();
                            } else if ("values".equals(n.getNodeName())) {
                                parseLogField(n);
                            }
                        }
                    }
                    if (propName != null && propValue != null) {
                        streamProperties.put(propName, propValue);
                    }
                }
            };

            Element root = doc.getDocumentElement();
            if (root != null && ROOT_ELEMENT_NAME.equals(root.getNodeName())) {
                NodeList nodeList = root.getChildNodes();
                if (nodeList != null) {
                    for (int i = 0; i < nodeList.getLength(); i++) {
                        Node node = nodeList.item(i);
                        if (node.getNodeType() == Node.ELEMENT_NODE) {
                            switch (node.getNodeName()) {
                                case STREAM_ELEMENT_NAME:
                                    parseStreamNode.accept(node);
                                    break;
                                case KAFKA_ELEMENT_NAME:
                                    parseProperty(node, kafkaProperties, "name", "value");
                                    break;
                                case HBASE_ELEMENT_NAME:
                                    parseProperty(node, hbaseProperties,"name", "value");
                                    break;
                                case RABBITMQ_ELEMENT_NAME:
                                    parseProperty(node, rabbitMqProperties, "name", "value");
                                default:
                                    break;
                            }
                        }
                    }
                } else {
                    throw new RuntimeException("配置文件格式错误!");
                }
            } else {
                throw new RuntimeException("配置文件格式错误!");
            }
        } catch (ParserConfigurationException e) {
            e.printStackTrace();
        } catch (SAXException e) {
            e.printStackTrace();
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    @SuppressWarnings("unchecked")
    private <E> void  parseProperty(Node parent, Map<String, E> map, String key, String val) {
        for (Node prop = parent.getFirstChild(); prop != null; prop = prop.getNextSibling()) {
            NodeList nodeList = prop.getChildNodes();
            String propName = null;
            E propValue = null;
            for (int i = 0; i < nodeList.getLength(); i++) {
                Node n = nodeList.item(i);
                if (n.getNodeType() == Node.ELEMENT_NODE) {
                    if (key.equals(n.getNodeName())) {
                        propName = n.getFirstChild().getNodeValue();
                    } else if (val.equals(n.getNodeName())) {
                        propValue = (E)n.getFirstChild().getNodeValue();
                    }
                }
            }
            if (propName != null && propValue != null) {
                map.put(propName, propValue);
            }
        }
    }

    private void parseLogField(Node parent) {
        for (Node prop = parent.getFirstChild(); prop != null; prop = prop.getNextSibling()) {
            NodeList nodeList = prop.getChildNodes();
            String field = null;
            String index = null;
            String cf = null;
            for (int i = 0; i < nodeList.getLength(); i++) {
                Node n = nodeList.item(i);
                if (n.getNodeType() == Node.ELEMENT_NODE) {
                    if ("field".equals(n.getNodeName())) {
                        field = n.getFirstChild().getNodeValue();
                    } else if ("index".equals(n.getNodeName())) {
                        index = n.getFirstChild().getNodeValue();
                    } else if ("cf".equals(n.getNodeName())) {
                        cf = n.getFirstChild().getNodeValue();
                    }
                }
            }
            if (field != null && index != null && cf != null) {
                logFields.put(Integer.valueOf(index), new HTableFiled(field, cf));
            }
        }
    }

    @Override
    public Map<Integer, HTableFiled> getLogFields() {
        return logFields;
    }

    @Override
    public String getHTableName() {
        return streamProperties.get("hbase.table.name");
    }

    @Override
    public Set<String> getLogType() {
        Set<String> set = new HashSet<>(Arrays.asList(COMMA.split(streamProperties.get("log.type"))));
        return set;
    }

    @Override
    public boolean saveErrorLog() {
        return Boolean.parseBoolean(streamProperties.getOrDefault("save.error.log", "true"));
    }

    @Override
    public boolean filterDev() {
        return Boolean.parseBoolean(streamProperties.getOrDefault("filter.dev", "true"));
    }

    @Override
    public List<String> getSubscribeTopics() {
        return Arrays.asList(COMMA.split(streamProperties.get("kafka.stream.topic")));
    }

    @Override
    public Map<String, Object> getKafkaConsumerConf() {
        return kafkaProperties;
    }

    @Override
    public Map<String, Object> getHBaseJobConf() {
        return hbaseProperties;
    }

    @Override
    public Map<String, String> getStreamConf() {
        return streamProperties;
    }

    @Override
    public int getAlertLevel() {
        return Integer.parseInt(streamProperties.getOrDefault("event.alert.level", "2"));
    }

    @Override
    public boolean autoReBroadcast() {
        return Boolean.parseBoolean(streamProperties.getOrDefault("enable.auto.broadcast", "false"));
    }

    @Override
    public int getReBroadcastInterval() {
        return Integer.parseInt(streamProperties.getOrDefault("broadcast.interval.ms", "3600000"));
    }

    @Override
    public Map<String, String> getRabbitMqConf() {
        return rabbitMqProperties;
    }
}
